package com.bsj.sipgateway.mq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;

/**
 * <p>
 * Title: MQ生产者
 * </p>
 * <p>
 * Description: 用于消息发布
 * </p>
 * <p>
 * Company: BSJ
 * </p>
 *
 * @author JIANG
 * @date 2017-12-25
 */
@Slf4j
public class RocketMqProducer {

	AtomicLong st_count = new AtomicLong();
	private DefaultMQProducer producer = null;

	public RocketMqProducer() {
		super();
	}

	public void start(String rocketmqAddr, String rocketmqPid) throws Exception {
		log.info("rocketmqAddr：" + rocketmqAddr + "，rocketmqPid：" + rocketmqPid);
		producer = new DefaultMQProducer(rocketmqPid);
		producer.setNamesrvAddr(rocketmqAddr);
		producer.start();
	}

	public void stop() {
		producer.shutdown();
	}

	/**
	 * 发送数据
	 *
	 * @param topic
	 *            会话
	 * @param tag
	 *            目标
	 * @param data
	 *            数据
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public void send(String topic, String tag, byte[] data)
			throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

		producer.send(new Message(topic, tag, data), new MessageQueueSelector() {
			// 每个节点轮着来
			public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
				// arg就是send的第3个参数
				long mqsCount = (Long) arg;
				return mqs.get((int) (mqsCount % mqs.size()));
			}
		}, st_count.incrementAndGet());
	}

	/**
	 * 发送数据
	 *
	 * @param topic
	 *            会话
	 * @param tag
	 *            目标
	 * @param data
	 *            数据
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public void send(String topic, String tag, byte[] data, long orderId)
			throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

		producer.send(new Message(topic, tag, data), new MessageQueueSelector() {
			// 每个节点轮着来
			public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
				// arg就是send的第3个参数
				long mqsCount = (Long) arg;
				return mqs.get((int) (mqsCount % mqs.size()));
			}
		}, orderId);
	}

	/**
	 * 发送数据
	 *
	 * @param topic
	 *            会话
	 * @param tag
	 *            目标
	 * @param data
	 *            数据
	 * @throws MQClientException
	 * @throws RemotingException
	 * @throws MQBrokerException
	 * @throws InterruptedException
	 */
	public SendResult send(String topic, String tag, byte[] data, int delayTimeLevel)
			throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
		Message message = new Message(topic, tag, data);
		if (0 != delayTimeLevel){
			message.setDelayTimeLevel(delayTimeLevel);
		}
		SendResult send = producer.send(message, new MessageQueueSelector() {
			// 每个节点轮着来
			@Override
			public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
				// arg就是send的第3个参数
				long mqsCount = (Long) arg;
				return mqs.get((int) (mqsCount % mqs.size()));
			}
		}, st_count.incrementAndGet());
		return send;
	}

}
